/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */

package cgl.granules.dataset;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Enumeration;
import java.util.LinkedList;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;

import cgl.granules.exception.CommunicationsException;
import cgl.granules.scheduler.StreamingService;
import cgl.granules.util.MarshallingUtil;
import cgl.narada.event.NBEvent;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.util.UUIDRetriever;

/**
 * This class is a streaming implementation of the dataset. Additionally, this
 * class implements the <code>StreamingAccess</code> interface which outlines
 * methods to access the data streams that are available.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class StreamDataset implements StreamingAccess, NBEventListener {

    private String datasetIdentifier = null;
    private String datasetDescription = null;
    private StreamingService streamingService;

    private EventConsumer consumer;

    private EventProducer producer;

    private boolean isInitialized = false;
    private boolean needsInitialization = true;

    private boolean isAvailable = false;

    private int datasetType = STREAMS;

    private AtomicLong lastModificationTime;

    private DatasetNotifier datasetNotifier;
    private String datasetNotifierBinding;

    private ConcurrentLinkedQueue<NBEvent> streamingQueue;

    private ConcurrentHashMap<String, Profile> profiles;

    private ConcurrentHashMap<String, Integer> inputStreams;

    private ConcurrentHashMap<String, Integer> outputStreams;

    private boolean closed = false;
    private boolean debug = true;
    private LinkedList<String> updates;
    private String moduleName = "StreamDataset: ";


    /**
     * Share the instance of the client service, so that multiple connections
     * are not created to the same broker.
     * 
     * @param datasetIdentifier
     *            The identifier of the dataset
     * @param datasetDescription
     *            The descriptor for this dataset.
     */
    public StreamDataset(String datasetIdentifier, String datasetDescription) {
        this.datasetIdentifier = datasetIdentifier;
        this.datasetDescription = datasetDescription;

        initializeDataStructuresAndStreaming();
    }


    private void initializeDataStructuresAndStreaming() {
        profiles = new ConcurrentHashMap<String, Profile>();
        streamingQueue = new ConcurrentLinkedQueue<NBEvent>();

        inputStreams = new ConcurrentHashMap<String, Integer>();
        outputStreams = new ConcurrentHashMap<String, Integer>();
        updates = new LinkedList<String>();

        lastModificationTime = new AtomicLong();
        UUIDRetriever retriever = UUIDRetriever.getInstance();
        datasetNotifierBinding = retriever.getRandomBasedUUIDAsString();
        datasetNotifier =
            new DatasetNotifier(datasetIdentifier, datasetNotifierBinding);
        streamingService = StreamingService.getInstance();
    }


    @Override
    public int addInputStream(String streamSynopsis, int synopsisType)
        throws DatasetException {
        inputStreams.putIfAbsent(streamSynopsis, synopsisType);
        updates.add(streamSynopsis);
        return inputStreams.size();
    }


    @Override
    public int removeInputStream(String streamSynopsis) throws DatasetException {
        if (!inputStreams.containsKey(streamSynopsis)) {
            String errorReport =
                "The stream (" + streamSynopsis
                    + ") was not previously registered.";
            throwDatasetException(errorReport);
        }
        updates.add(streamSynopsis);
        return inputStreams.size();
    }


    @Override
    public int addInputStreams(ConcurrentHashMap<String, Integer> streamsToAdd)
        throws DatasetException {

        if (streamsToAdd == null) {
            throw new NullPointerException("Streams to be added is NULL");
        }
        Enumeration<String> streams = streamsToAdd.keys();
        while (streams.hasMoreElements()) {
            String streamSynopsis = streams.nextElement();
            int synopsisType = streamsToAdd.get(streamSynopsis).intValue();
            addInputStream(streamSynopsis, synopsisType);
        }

        return inputStreams.size();
    }


    @Override
    public int removeInputStreams(
                                  ConcurrentHashMap<String, Integer> streamsToRemove)
        throws DatasetException {

        if (streamsToRemove == null) {
            throw new NullPointerException("Streams to be removed is NULL");
        }
        Enumeration<String> streams = streamsToRemove.keys();
        while (streams.hasMoreElements()) {
            String streamSynopsis = streams.nextElement();
            removeOutputStream(streamSynopsis);
        }

        return inputStreams.size();
    }


    @Override
    public boolean hasInputStreams() {
        if (inputStreams.isEmpty()) {
            return false;
        }
        return true;
    }


    @Override
    public int getNumberOfInputStreams() {
        return inputStreams.size();
    }


    @Override
    public ConcurrentHashMap<String, Integer> getInputStreams() {
        return inputStreams;
    }


    @Override
    public int addOutputStream(String streamSynopsis, int synopsisType)
        throws DatasetException {
        outputStreams.putIfAbsent(streamSynopsis, synopsisType);
        return outputStreams.size();
    }


    @Override
    public int removeOutputStream(String streamSynopsis)
        throws DatasetException {
        if (!outputStreams.containsKey(streamSynopsis)) {
            String errorReport =
                "The stream (" + streamSynopsis
                    + ") was not previously registered.";
            throwDatasetException(errorReport);
        }
        outputStreams.remove(streamSynopsis);
        return outputStreams.size();
    }


    @Override
    public int addOutputStreams(ConcurrentHashMap<String, Integer> streamsToAdd)
        throws DatasetException {

        if (streamsToAdd == null) {
            throw new NullPointerException("Streams to be added is NULL");
        }
        Enumeration<String> streams = streamsToAdd.keys();
        while (streams.hasMoreElements()) {
            String streamSynopsis = streams.nextElement();
            int synopsisType = streamsToAdd.get(streamSynopsis).intValue();
            addOutputStream(streamSynopsis, synopsisType);
        }

        return outputStreams.size();
    }


    @Override
    public int removeOutputStreams(
                                   ConcurrentHashMap<String, Integer> streamsToRemove)
        throws DatasetException {

        if (streamsToRemove == null) {
            throw new NullPointerException("Streams to be removed is NULL");
        }
        Enumeration<String> streams = streamsToRemove.keys();
        while (streams.hasMoreElements()) {
            String streamSynopsis = streams.nextElement();
            removeOutputStream(streamSynopsis);
        }

        return outputStreams.size();
    }


    @Override
    public boolean hasOutputStreams() {
        if (outputStreams.isEmpty()) {
            return false;
        }
        return true;
    }


    @Override
    public int getNumberOfOutputStreams() {
        return inputStreams.size();
    }


    @Override
    public ConcurrentHashMap<String, Integer> getOutputStreams() {
        return outputStreams;
    }


    @Override
    public void commitUpdates() throws DatasetException {
        while (!updates.isEmpty()) {
            String streamSynopsis = updates.poll();
            if (profiles.containsKey(streamSynopsis)) {
                System.out.println(moduleName + "Contains Profile for "
                    + streamSynopsis);
                Profile profile = profiles.get(streamSynopsis);
                unSubscribeConsumer(profile);
            } else {
                int synopsisType = inputStreams.get(streamSynopsis).intValue();
                subscribeConsumer(streamSynopsis, synopsisType);
            }
        }
    }


    @Override
    public boolean isDataAvailable() {
        boolean dataAvailable = true;
        if (streamingQueue.isEmpty()) {
            dataAvailable = false;
        }

        return dataAvailable;
    }


    @Override
    public NBEvent getStreamingData() {
        // lastModificationTime.set(System.currentTimeMillis());
        return streamingQueue.poll();
    }


    @Override
    public void writeStreamingData(NBEvent nbEvent) throws DatasetException {
        try {
            if (producer == null) {
                producer = streamingService.createProducer();
            }
            producer.publishEvent(nbEvent);

        } catch (ServiceException se) {
            String errorReport = se.toString();
            throwDatasetException(errorReport);
        } catch (CommunicationsException ce) {
            String errorReport = ce.toString();
            throwDatasetException(errorReport);
        }
    }


    @Override
    public void writeStreamingData(Object streamInfo, byte[] payload)
        throws DatasetException {
        int synopsisType = STRING_SYNOPSIS;
        writeStreamingData(synopsisType, streamInfo, payload);
    }


    @Override
    public void writeStreamingData(int synopsisType, Object streamInfo,
                                   byte[] payload) throws DatasetException {
        try {
            if (producer == null) {
                producer = streamingService.createProducer();
            }

            NBEvent nbEvent =
                producer.generateEvent(synopsisType, streamInfo, payload);
            producer.publishEvent(nbEvent);

        } catch (ServiceException se) {
            String errorReport = se.toString();
            throwDatasetException(errorReport);
        } catch (CommunicationsException ce) {
            String errorReport = ce.toString();
            throwDatasetException(errorReport);
        }

    }


    private void unsubscribe() throws DatasetException {
        String errorReport =
            moduleName + "Problems unsubscribing from some " + "streams\n";
        boolean error = false;

        Enumeration<Profile> e = profiles.elements();
        while (e.hasMoreElements()) {
            Profile profile = e.nextElement();
            try {
                consumer.unSubscribe(profile);
            } catch (ServiceException serEx) {
                error = true;
                errorReport += (profile.toString() + " unsubscribe failed\n");
            }
        }
        // datasetNotifier.datasetClosure(datasetNotifierBinding);

        if (error) {
            throwDatasetException(errorReport);
        }
    }


    @Override
    public boolean close() throws DatasetException {
        closed = true;

        return true;
    }


    @Override
    public DataAvailabilityNotifier getDataAvailabilityNotifier() {
        return datasetNotifier;
    }


    @Override
    public String getIdentifier() {
        return datasetIdentifier;
    }


    @Override
    public int getDatasetType() {
        return datasetType;
    }


    @Override
    public String getDescription() {
        return datasetDescription;
    }


    @Override
    public long getLastModificationTime() {
        return lastModificationTime.get();
    }


    @Override
    public boolean needsInitialization() {
        return needsInitialization;
    }


    @Override
    public boolean isInitialized() {
        return isInitialized;
    }


    @Override
    public boolean initializeDataset() throws DatasetException {
        commitUpdates();
        isInitialized = true;
        isAvailable = true;

        return isInitialized;
    }


    @Override
    public boolean isAvailable() {
        return isAvailable;
    }


    @Override
    public boolean supportsDataAvailabilityNotifications() {
        return true;
    }


    /**
     * Initialize the consumer (if need be), and subscribe to the stream
     * specified in the arguments.
     * 
     * @param synopsisTypeString
     * @param streamSynopsis
     * @throws DatasetException
     */
    private void subscribeConsumer(String streamSynopsis, int synopsisType)
        throws DatasetException {
        try {
            if (consumer == null) {
                consumer = streamingService.createConsumer(this);
            }

            Profile profile =
                streamingService.createProfile(streamSynopsis, synopsisType);

            consumer.subscribeTo(profile);
            profiles.put(streamSynopsis, profile);
            System.out.println("Subscribed to: " + streamSynopsis);
        } catch (CommunicationsException ce) {
            String errorReport = ce.toString();
            throwDatasetException(errorReport);
        } catch (ServiceException se) {
            String errorReport = se.toString();
            throwDatasetException(errorReport);
        }
    }


    private void unSubscribeConsumer(Profile profile) throws DatasetException {
        try {
            consumer.unSubscribe(profile);
        } catch (ServiceException se) {
            String errorReport =
                "Unsubcribe failed: " + profile.getSubscription() + "Reason = "
                    + se.toString() + "\n";
            throwDatasetException(errorReport);
        }
    }


    /**
     * This method will try to unsubscribe from as many streams as possible from
     * the list of streams that the dataset has subscribed to. All errors that
     * were encountered are reported when the method has finished the attempts.
     * 
     * @throws DatasetException
     */
    private void unSubscribeFromAllSubscribedStreams() throws DatasetException {
        String errorReport = moduleName + "Problems unsubscribing to: ";
        boolean reportError = false;
        String profileInfo = null;
        try {
            Enumeration<Profile> e = profiles.elements();
            while (!profiles.isEmpty()) {
                Profile profile = e.nextElement();
                profileInfo = profile.getSubscription().toString();
                consumer.unSubscribe(profile);
            }

        } catch (ServiceException se) {
            reportError = true;
            errorReport += (profileInfo + "Reason = " + se.toString() + "\n");
        }

        if (reportError) {
            throwDatasetException(errorReport);
        }
    }


    @Override
    public void onEvent(NBEvent nbEvent) {
        if (closed) {
            if (debug) {
                System.out.println("Dataset for ["
                    + nbEvent.getContentSynopsis() + "] is closed !");
            }
            return;
        }
        streamingQueue.add(nbEvent);
        // lastModificationTime.set(System.currentTimeMillis());

        datasetNotifier.dataAvailable(datasetNotifierBinding);
    }


    @Override
    public byte[] getBytes() throws IOException {
        MarshallingUtil marshaller = MarshallingUtil.getInstance();
        byte[] marshalledBytes = null;
        ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
        DataOutputStream dout = new DataOutputStream(baOutputStream);

        dout.writeInt(datasetType);
        marshaller.writeString(dout, datasetIdentifier);
        marshaller.writeString(dout, datasetDescription);

        dout.writeBoolean(needsInitialization);

        writeStreamInfo(inputStreams, dout, marshaller);
        writeStreamInfo(outputStreams, dout, marshaller);

        dout.flush();
        marshalledBytes = baOutputStream.toByteArray();
        return marshalledBytes;
    }


    private void writeStreamInfo(ConcurrentHashMap<String, Integer> streamInfo,
                                 DataOutputStream dout,
                                 MarshallingUtil marshaller) throws IOException {
        int numOfStreams = streamInfo.size();
        dout.writeInt(numOfStreams);

        Enumeration<String> streams = streamInfo.keys();
        int written = 0;
        while (streams.hasMoreElements()) {
            String synopsis = streams.nextElement();
            int synopsisType = streamInfo.get(synopsis).intValue();
            marshaller.writeString(dout, synopsis);
            dout.writeInt(synopsisType);
            written++;
        }

        if (written != numOfStreams) {
            throw new IOException(moduleName
                + "Streams information modified during getBytes()");
        }
    }


    public StreamDataset(byte[] marshalledBytes) throws IOException,
        DatasetException {
        ByteArrayInputStream baInputStream =
            new ByteArrayInputStream(marshalledBytes);
        DataInputStream din = new DataInputStream(baInputStream);
        MarshallingUtil marshaller = MarshallingUtil.getInstance();

        datasetType = din.readInt();
        datasetIdentifier = marshaller.readString(din);
        datasetDescription = marshaller.readString(din);

        needsInitialization = din.readBoolean();

        initializeDataStructuresAndStreaming();

        int numOfInputStreams = din.readInt();
        for (int i = 0; i < numOfInputStreams; i++) {
            String streamSynopsis = marshaller.readString(din);
            int synopsisType = din.readInt();
            addInputStream(streamSynopsis, synopsisType);
        }

        int numOfOutputStreams = din.readInt();
        for (int i = 0; i < numOfOutputStreams; i++) {
            String streamSynopsis = marshaller.readString(din);
            int synopsisType = din.readInt();
            addOutputStream(streamSynopsis, synopsisType);
        }

    }


    /**
     * Throws a data set exception based on the specified error report
     * 
     * @param errorReport
     * @throws DatasetException
     */
    private void throwDatasetException(String errorReport)
        throws DatasetException {
        DatasetException dse = new DatasetException(errorReport);
        throw dse;
    }
}
